/*
 *   Copyright 1999-2016 Asiainfo Technologies(China),Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.asiainfo.redis.scheduling;

import com.asiainfo.redis.entity.BusiDataChange;
import com.asiainfo.redis.service.BusiConfigService;
import com.asiainfo.redis.service.ScheduleService;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @version v 1.0 on 2016/8/11 15:27
 * @auther william.xu
 */
public class RefreshChangeBusiDataJob extends BaseJob {

    protected static final Logger logger = LoggerFactory.getLogger(RefreshChangeBusiDataJob.class);

    public void executeInternal(JobExecutionContext context) throws JobExecutionException {

        final ScheduleService scheduleService = applicationContext.getBean(ScheduleService.class);
        final BusiConfigService busiConfigService = applicationContext.getBean(BusiConfigService.class);

        //多线程同时处理
        ExecutorService es = Executors.newFixedThreadPool(ScheduleService.MAX_THREADS);
        List<BusiDataChange> list = scheduleService.getNotHandleBusiDataChangeList();
        for(final BusiDataChange busiDataChange : list){
            es.submit(new Callable<Object>() {
                @Override
                public Object call() throws Exception {

                    try {
                        busiDataChange.setStatus(ScheduleService.STATUS_RUNNING);
                        busiDataChange.setErrorMsg("");
                        busiDataChange.setUpdateTime(new java.util.Date());
                        scheduleService.saveBusiDataChange(busiDataChange);

                        busiConfigService.writeDataToRedis(busiDataChange.getBusiDataRuleId(),
                                busiDataChange.getSqlCondition(), busiDataChange.getOperFlag());
                        busiDataChange.setStatus(ScheduleService.STATUS_COMPLETED);
                        busiDataChange.setErrorMsg("");
                    }catch (Exception e){
                        logger.error("处理增量数据失败：id=" + busiDataChange.getId(), e);
                        busiDataChange.setErrorMsg(e.toString());
                        busiDataChange.setStatus(ScheduleService.STATUS_FAILED);
                    }
                    busiDataChange.setUpdateTime(new java.util.Date());
                    scheduleService.saveBusiDataChange(busiDataChange);

                    return null;
                }
            });

        }
        es.shutdown();
        //等待处理完
        try {
            es.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        }catch (Exception e){
            throw new JobExecutionException();
        }

    }


}
